package com.tuanzhang.zhw.rpc.client;

import com.tuanzhang.zhw.rpc.codec.RpcRequest;
import com.tuanzhang.zhw.rpc.codec.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 1.channelRegistered
 * 当通道注册完成后，Netty会调用fireChannelRegistered， 触发通道注册事件。通道会启动该入站操作的流水线处理，在通道注册过的入站处理器Handler的channelRegistered方法，会被调用到。
 * 2.channelActive
 * 当通道激活完成后，Netty会调用fireChannelActive， 触发通道激活事件。通道会启动该入站操作的流水线处理，在通道注册过的入站处理器Handler的channelActive方法，会被调用到。
 * 3.channelRead
 * 当通道缓冲区可读，Netty会调用fireChannelRead， 触发通道可读事件。通道会启动该入站操作的流水线处理，在通道注册过的入站处理器Handler的channelRead方法，会被调用到。
 * 4.channelReadComplete
 * 当通道缓冲区读完，Netty会调用fireChannelReadComplete， 触发通道读完事 件。通道会启动该入站操作的流水线处理，在通道注册过的入站处理器Handler的channelReadComplete方法，会被调用到。
 * 5.channelInactive
 * 当连接被断开或者不可用，Netty会调用fireChannelInactive， 触发连接不可用事 件。通道会启动对应的流水线处理，在通道注册过的入站处理器Handler的channelInactive方法，会被调用到。
 * 6.exceptionCaught
 * 当通道处理过程发生异常时，Netty会调用fireExceptionCaught，触发异常捕获事件。通道会启动异常捕获的流水线处理， 在通道注册过的处理器Handler的 exceptionCaught方法，会被调用到。注意，这个方法是在通道处理器中ChannelHandler定义的方法，入站处理器、出站处理器接口都继承到了该方法。
 * 完整的生命周期：
 * ChannelHandler中的回调方法的执行顺序为：
 * handlerAdded()→channelRegistered()→channelActive()→入站方法回调→channelInactive()→channelUnregistered()→handlerRemoved()。
 * 其中，读数据的入站回调为：channelRead()→channelReadComplete()；入站方法会多次调用，每一次有ByteBuf数据包入站都会调用到。除了两个入站回调方法外，其余的6个方法都和ChannelHandler的生命周期有关。
 */
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    /**
     * 缓存remotePeer: 通道连接的远端地址
     */
    private Channel channel;
    private SocketAddress remotePeer;

    /**
     * requestId-rpcFuture
     */
    private Map<String, RpcFuture> pendingRpcTable = new ConcurrentHashMap<>();


    public Channel getChannel() {
        return this.channel;
    }

    public SocketAddress getRemotePeer() {
        return remotePeer;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remotePeer = this.channel.remoteAddress();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcResponse rpcResponse) throws Exception {
        String requestId = rpcResponse.getRequestId();
        RpcFuture rpcFuture = pendingRpcTable.get(requestId);
        if (rpcFuture != null) {
            pendingRpcTable.remove(requestId);
            rpcFuture.done(rpcResponse);
        }
    }

    /**
     * 异步发送请求: Future模型: 可以支持Future#get方法, 通过其他线程获取返回结果
     * @param request
     * @return
     */
    public RpcFuture sendRequest(RpcRequest request){
        RpcFuture rpcFuture =new RpcFuture(request);
        pendingRpcTable.put(request.getRequestId(),rpcFuture);
        channel.writeAndFlush(request);
        return rpcFuture;
    }


    /**
     * Netty提供了一种主动关闭连接发的方式: 发送一个Unpooled.EMPTY_BUFFER, 这样ChannelFutureListener的Close事件就会监听到并关闭通道
     */
    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
}
